package com.bff.gaia.mix.api.xjoin.client.hbase.async;

import com.stumbleupon.async.Deferred;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.configuration.Configuration;
import com.bff.gaia.mix.api.functions.async.ResultFuture;
import com.bff.gaia.mix.api.xjoin.client.hbase.async.rowkeydealer.AbsRowKeyModeDealer;
import com.bff.gaia.mix.api.xjoin.client.hbase.async.rowkeydealer.PreRowKeyModeDealerDealer;
import com.bff.gaia.mix.api.xjoin.client.hbase.async.rowkeydealer.RowKeyEqualModeDealer;
import com.bff.gaia.mix.api.xjoin.client.hbase.core.table.HbaseSideTableInfo;
import com.bff.gaia.mix.api.xjoin.core.enums.ECacheContentType;
import com.bff.gaia.mix.api.xjoin.core.side.AsyncReqRow;
import com.bff.gaia.mix.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.mix.api.xjoin.core.side.JoinInfo;
import com.bff.gaia.mix.api.xjoin.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin.core.side.cache.CacheObj;
import com.bff.gaia.mix.api.xjoin.core.threadFactory.DTThreadFactory;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Maps;
import com.bff.gaia.types.Row;
import org.hbase.async.HBaseClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class HbaseAsyncReqRow extends AsyncReqRow {
	private static final long serialVersionUID = 2098635104857937717L;

	private static final Logger LOG = LoggerFactory.getLogger(HbaseAsyncReqRow.class);

	//match to the rule of netty3
	private static final int DEFAULT_BOSS_THREADS = 1;

	private static final int DEFAULT_IO_THREADS = Runtime.getRuntime().availableProcessors() * 2;

	private static final int DEFAULT_POOL_SIZE = DEFAULT_IO_THREADS + DEFAULT_BOSS_THREADS;

	private transient HBaseClient hBaseClient;

	private transient AbsRowKeyModeDealer rowKeyMode;

	private String tableName;

	private String[] colNames;

	public HbaseAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
		super(new HbaseAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));

		tableName = ((HbaseSideTableInfo) sideTableInfo).getTableName();
		colNames = ((HbaseSideTableInfo) sideTableInfo).getColumnRealNames();
	}


	@Override
	public void open(Configuration parameters) throws Exception {
		SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
		HbaseSideTableInfo hbaseSideTableInfo = (HbaseSideTableInfo) sideTableInfo;
		ExecutorService executorService = new ThreadPoolExecutor(DEFAULT_POOL_SIZE, DEFAULT_POOL_SIZE,
			0L, TimeUnit.MILLISECONDS,
			new LinkedBlockingQueue<>(), new DTThreadFactory("hbase-aysnc"));

		hBaseClient = new HBaseClient(hbaseSideTableInfo.getHost(), hbaseSideTableInfo.getParent(), executorService);

		try {
			Deferred deferred = hBaseClient.ensureTableExists(tableName)
				.addCallbacks(arg -> new CheckResult(true, ""), arg -> new CheckResult(false, arg.toString()));

			CheckResult result = (CheckResult) deferred.join();
			if (!result.isConnect()) {
				throw new RuntimeException(result.getExceptionMsg());
			}

		} catch (Exception e) {
			throw new RuntimeException("create hbase connection fail:", e);
		}

		HbaseAsyncSideInfo hbaseAsyncSideInfo = (HbaseAsyncSideInfo) sideInfo;
		if (hbaseSideTableInfo.isPreRowKey()) {
			rowKeyMode = new PreRowKeyModeDealerDealer(hbaseAsyncSideInfo.getColRefType(), colNames, hBaseClient,
				openCache(), sideInfo.getJoinType(), sideInfo.getOutFieldInfoList(),
				sideInfo.getInFieldIndex(), sideInfo.getSideFieldIndex());
		} else {
			rowKeyMode = new RowKeyEqualModeDealer(hbaseAsyncSideInfo.getColRefType(), colNames, hBaseClient,
				openCache(), sideInfo.getJoinType(), sideInfo.getOutFieldInfoList(),
				sideInfo.getInFieldIndex(), sideInfo.getSideFieldIndex());
		}
	}

	@Override
	public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
		Map<String, Object> refData = Maps.newHashMap();
		for (int i = 0; i < sideInfo.getEqualValIndex().size(); i++) {
			Integer conValIndex = sideInfo.getEqualValIndex().get(i);
			Object equalObj = input.getField(conValIndex);
			if (equalObj == null) {
				resultFuture.complete(null);
			}

			refData.put(sideInfo.getEqualFieldList().get(i), equalObj);
		}

		String rowKeyStr = ((HbaseAsyncSideInfo) sideInfo).getRowKeyBuilder().getRowKey(refData);

		//get from cache
		if (openCache()) {
			CacheObj val = getFromCache(rowKeyStr);
			if (val != null) {
				if (ECacheContentType.MissVal == val.getType()) {
					dealMissKey(input, resultFuture);
					return;
				} else if (ECacheContentType.SingleLine == val.getType()) {
					Row row = fillData(input, val);
					resultFuture.complete(Collections.singleton(row));
				} else if (ECacheContentType.MultiLine == val.getType()) {
					for (Object one : (List) val.getContent()) {
						Row row = fillData(input, one);
						resultFuture.complete(Collections.singleton(row));
					}
				}
				return;
			}
		}

		rowKeyMode.asyncGetData(tableName, rowKeyStr, input, resultFuture, sideInfo.getSideCache());
	}

	@Override
	public Row fillData(Row input, Object sideInput) {

		List<Object> sideInputList = (List<Object>) sideInput;
		Row row = new Row(sideInfo.getOutFieldInfoList().size());
		for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
			Object obj = input.getField(entry.getValue());
//			boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
			boolean isTimeIndicatorTypeInfo = false;
			if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
				obj = ((Timestamp) obj).getTime();
			}

			row.setField(entry.getKey(), obj);
		}

		for (Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()) {
			if (sideInputList == null) {
				row.setField(entry.getKey(), null);
			} else {
				row.setField(entry.getKey(), sideInputList.get(entry.getValue()));
			}
		}

		return row;
	}

	@Override
	public void close() throws Exception {
		super.close();
		hBaseClient.shutdown();
	}


	class CheckResult {

		private boolean connect;

		private String exceptionMsg;

		CheckResult(boolean connect, String msg) {
			this.connect = connect;
			this.exceptionMsg = msg;
		}

		public boolean isConnect() {
			return connect;
		}

		public void setConnect(boolean connect) {
			this.connect = connect;
		}

		public String getExceptionMsg() {
			return exceptionMsg;
		}

		public void setExceptionMsg(String exceptionMsg) {
			this.exceptionMsg = exceptionMsg;
		}
	}
}